package com.atguigu.udmp.stream.process;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class OrderJoinProcess {
    public static DataStream<JSONObject> process(DataStream<JSONObject> jsonObjectDataStream){

        //1 定义分流标签
        OutputTag<JSONObject> orderInfoTag=  new OutputTag<JSONObject>("order_info"){};
        OutputTag<JSONObject> orderDetailTag=  new OutputTag<JSONObject>("order_detail"){};

        //2 从主流 中分出两个子流
        SingleOutputStreamOperator<JSONObject> mainStream = jsonObjectDataStream.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
 //               System.out.println("jsonObject = " + jsonObject);
                String table = jsonObject.getString("table");
                if (table != null && table.equals("order_info")) {
                    collector.collect(jsonObject);
                    context.output(orderInfoTag, jsonObject);
                } else if (table != null && table.equals("order_detail")) {
                    context.output(orderDetailTag, jsonObject);
                } else {
                    collector.collect(jsonObject);
                }

            }
        });
        SideOutputDataStream<JSONObject> orderInfoStream = mainStream.getSideOutput(orderInfoTag);
        SideOutputDataStream<JSONObject> orderDetailStream = mainStream.getSideOutput(orderDetailTag);


        //3 分别设定水位线
        SingleOutputStreamOperator<JSONObject> orderInfoWmStream = orderInfoStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts")*1000L;
            }
        }));

        SingleOutputStreamOperator<JSONObject> orderDetailWmStream = orderDetailStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts")*1000L;
            }
        }));

        //4 分别设定key
        KeyedStream<JSONObject, String> orderInfoKeyedStream = orderInfoWmStream.keyBy(jsonObject -> jsonObject.getJSONObject("data").getString("id"));
        KeyedStream<JSONObject, String> orderDetailKeyedStream = orderDetailWmStream.keyBy(jsonObject -> jsonObject.getJSONObject("data").getString("order_id"));



    //      orderInfoKeyedStream.print("orderInfoKeyed::::");
    //     orderDetailKeyedStream.print("orderDetailKeyed::::");

        //4 利用intervalJoin关联
        SingleOutputStreamOperator<JSONObject> orderDetailConnectedStream = orderInfoKeyedStream.intervalJoin(orderDetailKeyedStream).between(Time.seconds(-1000),Time.seconds(20*60*1000)).process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject orderInfoJsonObj, JSONObject orderDetailJsonObj, ProcessJoinFunction<JSONObject, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                JSONObject orderDetailDataJsonObj = orderDetailJsonObj.getJSONObject("data");
                JSONObject orderInfoDataJsonObj = orderInfoJsonObj.getJSONObject("data");
                if (orderDetailDataJsonObj != null) {
                    orderDetailDataJsonObj.put("user_id", orderInfoDataJsonObj.getString("user_id"));
                    orderDetailDataJsonObj.put("order_status", orderInfoDataJsonObj.getString("order_status"));
                    orderDetailDataJsonObj.put("operate_time",orderInfoDataJsonObj.getString("operate_time"));
                }
                collector.collect(orderDetailJsonObj);

            }
        });
    //    orderDetailConnectedStream.print("orderDetailConnectedStream::");
        //5  再合并到主流中
        DataStream<JSONObject> mainNewStream = mainStream.union(orderDetailConnectedStream);
        return mainNewStream;
    }
}
